/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.news.server.util;

import cn.ac.iie.di.ban.news.server.exception.NoSuchDataTypeException;
import cn.ac.iie.di.ban.news.server.runtime.RuntimeEnv;
import cn.ac.iie.di.ban.news.server.runtime.RuntimeEnvException;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.RESessionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author Black-Laptop
 */
public class DataSender {

    private static final Logger logger = LoggerFactory.getLogger(DataSender.class);

    private static final String TYPE_FIELD_NAME = "msg_type";
    private static final String DATA_KEY_FIELD_NAME = "key";
    private static final String DATA_FIELD_NAME = "record_list";
    private static final String GT_FIELD_NAME = "gt";

    private static final HashSet<String> GLOBAL_FIELDS;

    static {
        GLOBAL_FIELDS = new HashSet<>();
        GLOBAL_FIELDS.add(TYPE_FIELD_NAME);
        GLOBAL_FIELDS.add(DATA_KEY_FIELD_NAME);
        GLOBAL_FIELDS.add(DATA_FIELD_NAME);
        GLOBAL_FIELDS.add("total_pushed_num");
    }

    private final HashMap<String, SenderPool> senderpools;
    private final HashMap<String, Map<String, REFieldType>> typeMaps;

    public DataSender() {
        this.senderpools = new HashMap<>();
        this.typeMaps = new HashMap<>();
    }

    public boolean regSender(String type, String namesrv, String topic, Map<String, REFieldType> typeMap, int maxParrallel) throws REConnectionException {
        if (!typeMap.containsKey(GT_FIELD_NAME)) {
            typeMap.put(GT_FIELD_NAME, REFieldType.Long);
        }
        //exist
        if (senderpools.containsKey(type)) {
            return false;
        }
        //not exist
        senderpools.put(type, new SenderPool(namesrv, topic, typeMap, maxParrallel));
        typeMaps.put(type, typeMap);
        return true;
    }

    public boolean regSender(String type, String namesrv, String topic, Map<String, REFieldType> typeMap) throws REConnectionException {
        if (!typeMap.containsKey(GT_FIELD_NAME)) {
            typeMap.put(GT_FIELD_NAME, REFieldType.Long);
        }
        //exist
        if (senderpools.containsKey(type)) {
            return false;
        }
        //not exist
        senderpools.put(type, new SenderPool(namesrv, topic, typeMap));
        typeMaps.put(type, typeMap);
        return true;
    }

    public void sendMessage(JSONObject obj, long gt) throws RESessionException, REConnectionException, NoSuchDataTypeException {
        logger.debug("start send msg to rmq, data: " + obj.toJSONString());
        //confirm type
        String type;
        try {
            type = obj.getString(TYPE_FIELD_NAME);
        } catch (Exception e) {
            return;
        }
        if (type == null) {
            return;
        }
        if (!senderpools.containsKey(type)) {
//			throw new NoSuchDataTypeException("no data type \"" + type + "\".");
            logger.error("No rmq sender can be used for data type " + type + " ,maybe data type not registed!");
            return;
        }
        //borrow sender
        Map<String, REFieldType> typeMap = typeMaps.get(type);
        RESendSession sender = senderpools.get(type).Borrow();
        //user desc
        HashMap<String, String> userDesc = new HashMap<>();
        for (String col : GLOBAL_FIELDS) {
            if (typeMap.containsKey(col)) {
                userDesc.put(col, obj.getString(col));
            }
            userDesc.put(GT_FIELD_NAME, "" + gt);
        }
        if (userDesc.isEmpty()) {
        } else {
            sender.setUserDesc(userDesc);
        }
        //data
        try {
            for (Object rec : obj.getJSONArray(DATA_FIELD_NAME)) {
                for (String col : GLOBAL_FIELDS) {
                    if (typeMap.containsKey(col)) {
                        String val = obj.getString(col);
                        switch (typeMap.get(col)) {
                            case Strings:
                                JSONArray a = JSONArray.parseArray(val);
                                String[] strs = new String[a.size()];
                                for (int i = 0; i < a.size(); ++i) {
                                    strs[i] = a.getString(i);
                                }
                                sender.setStrings(col, strs);
                                break;
                            case String:
                                sender.setString(col, val);
                                break;
                            case Int:
                                sender.setInt(col, Integer.parseInt(val));
                                break;
                            case Long:
                                sender.setLong(col, Long.parseLong(val));
                                break;
                        }
                    }
                }
                for (Map.Entry<String, Object> o : ((JSONObject) rec).entrySet()) {
                    String col = o.getKey();
                    if (typeMap.containsKey(col)) {
                        Object v = o.getValue();
                        if (v == null) {
                        } else {
                            String val = v.toString();
                            switch (typeMap.get(col)) {
                                case Strings:
                                    JSONArray a = JSONArray.parseArray(val);
                                    String[] strs = new String[a.size()];
                                    for (int i = 0; i < a.size(); ++i) {
                                        strs[i] = a.getString(i);
                                    }
                                    sender.setStrings(col, strs);
                                    break;
                                case String:
                                    sender.setString(col, val);
                                    break;
                                case Int:
                                    sender.setInt(col, Integer.parseInt(val));
                                    break;
                                case Long:
                                    sender.setLong(col, Long.parseLong(val));
                                    break;
                            }
                        }
                    }
                }
                sender.setLong(GT_FIELD_NAME, gt);
                try {
                    sender.add();
                } catch (Exception e) {
                    while (true) {
                        try {
                            sender.flush();
                            break;
                        } catch (Exception ex) {
                        }
                    }
                    sender.add();
                }
            }
            sender.flush();
            logger.debug("send to rmq succeeded.");
        } finally {
            senderpools.get(type).Return(sender);
        }
    }

    public void sendMessage(JSONObject obj) throws RESessionException, REConnectionException, NoSuchDataTypeException {
        sendMessage(obj, System.currentTimeMillis());
    }

    public static void main(String[] args) {
        Object o = "";
        JSONArray a = JSONArray.parseArray(o.toString());
        String[] strs = new String[a.size()];
        for (int i = 0; i < a.size(); ++i) {
            strs[i] = a.getString(i);
        }
        System.out.println("aaa"+Arrays.asList(strs));
    }

}
